package org.example.service;

import com.alipay.remoting.exception.CodecException;
import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import lombok.extern.slf4j.Slf4j;
import org.example.CommonMsg;
import org.example.MsgConstants;
import org.example.command.OrderCmd;
import org.example.conter.CounterPacket;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 用来和网关通信，这里的网关是指交易所的网关。
 * 负载柜台和gateway的通信，
 */
@Service
@Slf4j
public class GatewayService {

    @Value("${server.gateway.ip}")
    String ip;

    @Value("${server.gateway.port}")
    int port;

    private Vertx vertx;

    private volatile NetSocket socket;

    private final BlockingQueue<Buffer> cache = new LinkedBlockingQueue<>();


    @PostConstruct
    public void init() {
        vertx = Vertx.vertx();
        vertx.createNetClient().connect(port,ip,new ClientHandler());

        Executors.newSingleThreadExecutor().execute(
                () -> {
                    while(true) {
                        Buffer buffer = null;
                        try {
                            System.out.println("--------------->"+ cache.size());
                            buffer = cache.poll(1, TimeUnit.SECONDS);
                            if(buffer != null && buffer.length() > 0  && socket != null) {
                                System.out.println("===> try to write.....");
                                CommonMsg commonMsg = CommonMsg.decodeFromBufferNoCsCheck(buffer);
                                System.out.println("check sum = " + commonMsg.getChecksum());
                                socket.write(buffer);
                                System.out.println("===> over write.....");
                            }
                        } catch (InterruptedException e) {
                            log.error("msg send fail");
                        }
                        try {

                            Thread.sleep(10);
                        } catch (InterruptedException e) {
                            log.error("===sleep wake up");
                        }
                    }
                }
        );
    }

    public void submitOrder(OrderCmd orderCmd) throws CodecException {
        byte[] serialize = orderCmd.serialize();
        CommonMsg msg = new CommonMsg();
        msg.setBodyLength(serialize.length);
        msg.setMsgSrc((short) 1);//TODO 先随便写下
        msg.setMsgDst((short) 1);//TODO 后面改
        msg.setMsgType(MsgConstants.COUNTER_NEW_ORDER);
        msg.setStatus(MsgConstants.NORMAL);
        msg.setMsgNo(2);//TODO 随便写的，后面改
        msg.setBody(serialize);
        Buffer data = CounterPacket.writeToBuffer(msg);
        cache.offer(data);
        System.out.println("缓存大小="+cache.size());
        System.out.println("现在缓存大小="+cache.size());
    }

    private class ClientHandler implements Handler<AsyncResult<NetSocket>> {

        private void reconnect() {
            vertx.setTimer(5000, r -> {
                vertx.createNetClient().connect(port, ip, new ClientHandler());
            });
        }

        @Override
        public void handle(AsyncResult<NetSocket> netSocketAsyncResult) {

            if(netSocketAsyncResult.succeeded()) {
                System.out.println("start tcp client");
                socket = netSocketAsyncResult.result();
                socket.closeHandler(close -> {
                    System.out.println("tcp closed, try to reconnect");
                    reconnect();
                });

                socket.exceptionHandler(ex -> {
                    log.error("exception on connection");
                });
            }else {
                log.error(netSocketAsyncResult.toString());
            }
        }
    }
}
